package org.ws.task.zookeeper.listener;

import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.ws.task.listener.*;
import org.ws.task.listener.zk.CompleteNodeListener;
import org.ws.task.listener.zk.LeaderNodeListener;
import org.ws.task.listener.zk.NodeListener;
import org.ws.task.listener.zk.TaskNodeListener;
import org.ws.task.listener.zk.WorkerNodeListener;
import org.ws.task.model.NodeMetadata;
import org.ws.task.model.TaskConfig;

import java.util.List;

public class NodeListenerManager implements RegisterListener<NodeListener>, CompleteNodeListener {

    private static Logger log = LoggerFactory.getLogger(NodeListenerManager.class.getName());

    private final List<LeaderNodeListener> leaderNodeListeners;
    private final List<WorkerNodeListener> workerNodeListeners;
    private final List<TaskNodeListener> taskNodeListeners;

    public NodeListenerManager(){
        this.leaderNodeListeners = Lists.newLinkedList();
        this.workerNodeListeners = Lists.newLinkedList();
        this.taskNodeListeners = Lists.newLinkedList();
    }

    @Override
    public void registerListener(NodeListener listener) {
        boolean matched = false;
        if(listener instanceof LeaderNodeListener){
            leaderNodeListeners.add((LeaderNodeListener) listener);
            matched = true;
        }
        if(listener instanceof TaskNodeListener){
            taskNodeListeners.add((TaskNodeListener) listener);
            matched = true;
        }
        if(listener instanceof WorkerNodeListener){
            workerNodeListeners.add((WorkerNodeListener) listener);
            matched = true;
        }
        if(!matched) {
            log.warn("listener:[{}] can not match.",listener.getClass().getName());
        }
    }

    @Override
    public void becomeLeader(final NodeMetadata nodeMetadata) {
        this.leaderNodeListeners.forEach(l -> l.becomeLeader(nodeMetadata));
    }

    @Override
    public void notLeader(NodeMetadata nodeMetadata) {
        this.leaderNodeListeners.forEach(l -> l.notLeader(nodeMetadata));
    }

    @Override
    public void changeLeaderNode(final NodeMetadata nodeMetadata) {
        this.leaderNodeListeners.forEach(l -> l.changeLeaderNode(nodeMetadata));
    }

    @Override
    public void removeLeaderNode(final NodeMetadata nodeMetadata) {
        this.leaderNodeListeners.forEach(l -> l.removeLeaderNode(nodeMetadata));
    }

    @Override
    public void updateTaskConfig(final TaskConfig taskConfig) {
        this.taskNodeListeners.forEach(l->l.updateTaskConfig(taskConfig));
    }

    @Override
    public void removeTaskConfig(final TaskConfig taskConfig) {
        this.taskNodeListeners.forEach(l->l.removeTaskConfig(taskConfig));
    }

    @Override
    public void updateWorkerNode(final NodeMetadata nodeMetadata) {
        this.workerNodeListeners.forEach(l->l.updateWorkerNode(nodeMetadata));
    }

    @Override
    public void removeWorkerNode(final NodeMetadata nodeMetadata) {
        this.workerNodeListeners.forEach(l->l.removeWorkerNode(nodeMetadata));
    }
}
